package com.heima;

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.*;

@Slf4j
@RestController
public class RedissonController implements InitializingBean {

    private static final String QUEUE_NAME = "delay_queue_demo";

    private ExecutorService taskMsgService = Executors.newSingleThreadExecutor();

    private ExecutorService businessService = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors()*2,
            120L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(1000),
            new ThreadPoolExecutor.CallerRunsPolicy());

    @Autowired
    private RedissonClient redissonClient;

    @GetMapping("/put/{value}/{seconds}")
    public String put(@PathVariable("value")String value,
                      @PathVariable("seconds")Long seconds){
        RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(QUEUE_NAME);
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        delayedQueue.offer(value, seconds, TimeUnit.SECONDS);
        log.info("放入延迟任务, value={}, seconds={}", value, seconds);
        return value + "," + seconds;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        taskMsgService.execute(new Runnable() {
            @Override
            public void run() {
                RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(QUEUE_NAME);
                try{
                    while(true){
                        String value = blockingDeque.take();
                        // 业务逻辑处理
                        log.info("收到延迟任务： value={}", value);
                        businessService.execute(new Runnable() {
                            @Override
                            public void run() {
                                log.info("业务逻辑处理： value={}", value);
                            }
                        });
                    }
                }catch(Exception e){
                    e.printStackTrace();
                }
            }
        });
    }
}
